package server

import (
	"errors"
	"fmt"
	"log"
	"net/http"
	"sync"
	"time"

	"github.com/coder/websocket"
	"github.com/xkgo/lcs/binary"
)

// Server 是长连接服务的主要实现
type Server struct {
	// 客户端连接映射表，key是客户端ID
	clients     map[string]*Client
	clientsLock sync.RWMutex

	// 本地订阅映射表，用于快速访问，key是topic，value是订阅该topic的客户端ID集合
	subscriptions     map[string]map[string]struct{}
	subscriptionsLock sync.RWMutex

	// 会话存储
	sessionStore SessionStore

	// 服务器ID
	serverID string

	// 服务器地址
	serverAddr string

	// RPC客户端连接池
	rpcClientPool *RPCClientPool

	// 配置选项
	options ServerOptions
}

// ServerOptions 服务器配置选项
type ServerOptions struct {
	// 读取消息超时时间
	ReadTimeout time.Duration
	// 写入消息超时时间
	WriteTimeout time.Duration
	// 最大消息大小
	MaxMessageSize int64
	// 服务器ID
	ServerID string
	// 服务器地址，用于RPC调用，格式为 "IP:端口"
	ServerAddr string
	// 会话存储类型: "memory" 或 "redis"
	SessionStoreType string
	// Redis配置，当SessionStoreType为"redis"时使用
	RedisOptions RedisSessionOptions
	// 是否启用token验证
	EnableTokenValidation bool
	// token验证器
	TokenValidator TokenValidator
}

// DefaultServerOptions 返回默认的服务器配置
func DefaultServerOptions() ServerOptions {
	return ServerOptions{
		ReadTimeout:           time.Minute,
		WriteTimeout:          time.Second * 10,
		MaxMessageSize:        1024 * 1024, // 1MB
		ServerID:              "server1",
		ServerAddr:            "localhost:8081",
		SessionStoreType:      "memory",
		RedisOptions:          DefaultRedisSessionOptions(),
		EnableTokenValidation: false,
		TokenValidator:        &NoopTokenValidator{},
	}
}

// NewServer 创建一个新的长连接服务器
func NewServer(options ...ServerOptions) *Server {
	opts := DefaultServerOptions()
	if len(options) > 0 {
		opts = options[0]
	}

	// 创建会话存储
	var sessionStore SessionStore
	switch opts.SessionStoreType {
	case "redis":
		sessionStore = NewRedisSessionStore(opts.RedisOptions)
		log.Printf("Using Redis session store at %s", opts.RedisOptions.Addr)
	default:
		sessionStore = NewMemorySessionStore()
		log.Printf("Using memory session store")
	}

	// 创建RPC客户端连接池
	rpcClientPool := NewRPCClientPool()

	// 创建服务器实例
	server := &Server{
		clients:       make(map[string]*Client),
		subscriptions: make(map[string]map[string]struct{}),
		sessionStore:  sessionStore,
		serverID:      opts.ServerID,
		serverAddr:    opts.ServerAddr,
		rpcClientPool: rpcClientPool,
		options:       opts,
	}

	// 启动RPC服务器
	if err := StartRPCServer(server, opts.ServerAddr); err != nil {
		log.Printf("Failed to start RPC server: %v", err)
	}

	// 如果使用Redis存储，更新Redis选项中的服务器地址
	if redisStore, ok := sessionStore.(*RedisSessionStore); ok {
		// 更新服务器地址
		redisStore.serverAddr = opts.ServerAddr
	}

	return server
}

// ServeHTTP 实现http.Handler接口，处理WebSocket连接请求
func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	// 从请求中获取用户ID
	uid := r.URL.Query().Get("uid")
	if uid == "" {
		http.Error(w, "Missing uid parameter", http.StatusBadRequest)
		return
	}

	// 如果启用了token验证，则验证token
	if s.options.EnableTokenValidation {
		// 获取appID和token
		appID := r.URL.Query().Get("appid")
		if appID == "" {
			http.Error(w, "Missing appid parameter", http.StatusBadRequest)
			return
		}

		token := r.URL.Query().Get("token")
		if token == "" {
			http.Error(w, "Missing token parameter", http.StatusBadRequest)
			return
		}

		// 验证token
		if err := s.options.TokenValidator.ValidateToken(appID, uid, token); err != nil {
			log.Printf("Token validation failed for appID=%s, uid=%s: %v", appID, uid, err)
			http.Error(w, "Invalid token", http.StatusUnauthorized)
			return
		}

		log.Printf("Token validation successful for appID=%s, uid=%s", appID, uid)
	}

	// 建立WebSocket连接
	c, err := websocket.Accept(w, r, &websocket.AcceptOptions{
		CompressionMode: websocket.CompressionDisabled,
	})
	if err != nil {
		log.Printf("Failed to accept websocket connection: %v", err)
		return
	}

	// 创建客户端对象
	client := NewClient(uid, c, s)

	// 注册客户端
	s.registerClient(client)

	// 处理客户端连接
	client.Start()
}

// registerClient 注册一个新的客户端连接
func (s *Server) registerClient(client *Client) {
	s.clientsLock.Lock()
	defer s.clientsLock.Unlock()

	// 如果已存在相同UID的客户端，先关闭旧连接
	if oldClient, exists := s.clients[client.ID]; exists {
		oldClient.Close()
	}

	// 将客户端添加到本地映射表
	s.clients[client.ID] = client

	// 将客户端信息保存到会话存储
	if err := s.sessionStore.SaveClient(client.ID, s.serverID); err != nil {
		log.Printf("Failed to save client to session store: %v", err)
	}

	log.Printf("Client registered: %s", client.ID)
}

// unregisterClient 注销一个客户端连接
func (s *Server) unregisterClient(clientID string) {
	// 从客户端映射表中移除
	s.clientsLock.Lock()
	delete(s.clients, clientID)
	s.clientsLock.Unlock()

	// 从会话存储中移除客户端
	if err := s.sessionStore.RemoveClient(clientID); err != nil {
		log.Printf("Failed to remove client from session store: %v", err)
	}

	// 获取客户端订阅的所有主题
	topics, err := s.getClientTopics(clientID)
	if err != nil {
		log.Printf("Failed to get client topics: %v", err)
	}

	// 从所有订阅中移除该客户端
	for _, topic := range topics {
		if err := s.sessionStore.RemoveSubscription(clientID, topic); err != nil {
			log.Printf("Failed to remove subscription from session store: %v", err)
		}
	}

	log.Printf("Client unregistered: %s", clientID)
}

// getClientTopics 获取客户端订阅的所有主题
func (s *Server) getClientTopics(clientID string) ([]string, error) {
	// 这里我们需要遍历所有主题，查找客户端订阅的主题
	// 在实际应用中，可以优化这个过程，例如在Redis中为每个客户端维护一个订阅列表

	// 在内存存储中实现
	if memStore, ok := s.sessionStore.(*MemorySessionStore); ok {
		topics := []string{}
		for topic, subscribers := range memStore.subscriptions {
			if _, exists := subscribers[clientID]; exists {
				topics = append(topics, topic)
			}
		}
		return topics, nil
	}

	// 对于Redis存储，我们需要遍历所有主题
	// 这里的实现不是最优的，在实际应用中应该优化
	// 返回空列表，因为我们没有简单的方法获取客户端的所有订阅
	return []string{}, nil
}

// Subscribe 订阅指定的topic
func (s *Server) Subscribe(clientID string, topic string) {
	// 将订阅信息保存到会话存储
	if err := s.sessionStore.SaveSubscription(clientID, topic); err != nil {
		log.Printf("Failed to save subscription to session store: %v", err)
	}

	// 同时保存到本地内存中，便于快速访问
	s.subscriptionsLock.Lock()
	defer s.subscriptionsLock.Unlock()

	// 如果topic不存在，创建一个新的订阅者集合
	if _, exists := s.subscriptions[topic]; !exists {
		s.subscriptions[topic] = make(map[string]struct{})
	}

	// 添加客户端到订阅者集合
	s.subscriptions[topic][clientID] = struct{}{}
	log.Printf("Client %s subscribed to topic: %s", clientID, topic)
}

// Unsubscribe 取消订阅指定的topic
func (s *Server) Unsubscribe(clientID string, topic string) {
	// 从会话存储中移除订阅信息
	if err := s.sessionStore.RemoveSubscription(clientID, topic); err != nil {
		log.Printf("Failed to remove subscription from session store: %v", err)
	}

	// 同时从本地内存中移除
	s.subscriptionsLock.Lock()
	defer s.subscriptionsLock.Unlock()

	// 如果topic存在，从订阅者集合中移除客户端
	if subscribers, exists := s.subscriptions[topic]; exists {
		delete(subscribers, clientID)
		// 如果该topic没有订阅者了，删除该topic
		if len(subscribers) == 0 {
			delete(s.subscriptions, topic)
		}
		log.Printf("Client %s unsubscribed from topic: %s", clientID, topic)
	}
}

// Publish 向指定topic的所有订阅者发布消息
func (s *Server) Publish(topic string, message []byte) {
	// 从会话存储中获取订阅者
	subsFromStore, err := s.sessionStore.GetTopicSubscribers(topic)
	if err != nil {
		log.Printf("Failed to get subscribers from session store: %v", err)
	}

	// 从本地内存中获取订阅者
	s.subscriptionsLock.RLock()
	subscribers, exists := s.subscriptions[topic]
	s.subscriptionsLock.RUnlock()

	// 合并订阅者列表
	subsMap := make(map[string]struct{})

	// 添加从会话存储中获取的订阅者
	for _, clientID := range subsFromStore {
		subsMap[clientID] = struct{}{}
	}

	// 添加从本地内存中获取的订阅者
	if exists {
		for clientID := range subscribers {
			subsMap[clientID] = struct{}{}
		}
	}

	// 如果没有订阅者，直接返回
	if len(subsMap) == 0 {
		return
	}

	// 复制订阅者列表，避免在发送过程中锁定
	clientIDs := make([]string, 0, len(subsMap))
	for clientID := range subsMap {
		clientIDs = append(clientIDs, clientID)
	}

	// 向所有订阅者发送消息
	for _, clientID := range clientIDs {
		s.SendToClient(clientID, message)
	}

	log.Printf("Published message to topic %s, %d subscribers", topic, len(clientIDs))
}

// sendToLocalClient 向本地客户端发送消息
func (s *Server) sendToLocalClient(clientID string, message []byte) error {
	// 检查本地客户端
	s.clientsLock.RLock()
	client, exists := s.clients[clientID]
	s.clientsLock.RUnlock()

	if !exists {
		return errors.New("client not found locally")
	}

	// 发送消息
	return client.Send(message)
}

// SendToClient 向指定客户端发送消息
func (s *Server) SendToClient(clientID string, message []byte) error {
	// 先尝试发送到本地客户端
	err := s.sendToLocalClient(clientID, message)
	if err == nil {
		// 成功发送到本地客户端
		return nil
	}

	// 如果客户端不在本地，检查它是否在其他服务器上
	if redisStore, ok := s.sessionStore.(*RedisSessionStore); ok {
		// 获取客户端信息
		clientInfo, err := redisStore.GetClientInfo(clientID)
		if err != nil {
			return fmt.Errorf("client not found: %v", err)
		}

		// 如果客户端在其他服务器上，通过RPC调用转发消息
		if clientInfo.ServerID != s.serverID && clientInfo.ServerAddr != "" {
			// 检查目标服务器是否就是当前服务器
			if clientInfo.ServerAddr == s.serverAddr {
				// 如果是当前服务器，直接尝试发送到本地客户端
				// 这种情况可能发生在客户端断开连接后重新连接到同一个服务器，但Redis中的信息还没有更新
				log.Printf("Client %s is registered on this server with different ID, trying local send", clientID)
				return s.sendToLocalClient(clientID, message)
			}

			log.Printf("Forwarding message to client %s on server %s (%s)", clientID, clientInfo.ServerID, clientInfo.ServerAddr)

			// 获取RPC客户端
			client, err := s.rpcClientPool.GetClient(clientInfo.ServerAddr)
			if err != nil {
				return fmt.Errorf("failed to get RPC client: %v", err)
			}

			// 发送消息
			return client.SendMessage(clientID, message)
		}
	}

	// 如果不是Redis存储或客户端信息不完整，无法转发消息
	return errors.New("client not found or message forwarding is not supported")
}

// SendToClients 向多个客户端发送相同的消息
func (s *Server) SendToClients(clientIDs []string, message []byte) {
	for _, clientID := range clientIDs {
		if err := s.SendToClient(clientID, message); err != nil {
			log.Printf("Failed to send message to client %s: %v", clientID, err)
		}
	}
}

// GetClient 获取指定ID的客户端
func (s *Server) GetClient(clientID string) (*Client, bool) {
	s.clientsLock.RLock()
	defer s.clientsLock.RUnlock()
	client, exists := s.clients[clientID]
	return client, exists
}

// GetAllClients 获取所有客户端
func (s *Server) GetAllClients() []*Client {
	s.clientsLock.RLock()
	defer s.clientsLock.RUnlock()

	clients := make([]*Client, 0, len(s.clients))
	for _, client := range s.clients {
		clients = append(clients, client)
	}
	return clients
}

// GetClientCount 获取当前连接的客户端数量
func (s *Server) GetClientCount() int {
	s.clientsLock.RLock()
	defer s.clientsLock.RUnlock()
	return len(s.clients)
}

// GetSubscriberCount 获取指定topic的订阅者数量
func (s *Server) GetSubscriberCount(topic string) int {
	s.subscriptionsLock.RLock()
	defer s.subscriptionsLock.RUnlock()

	if subscribers, exists := s.subscriptions[topic]; exists {
		return len(subscribers)
	}
	return 0
}

// Close 关闭服务器，释放资源
func (s *Server) Close() error {
	// 关闭RPC客户端连接池
	if s.rpcClientPool != nil {
		s.rpcClientPool.Close()
	}

	// 关闭会话存储
	if s.sessionStore != nil {
		return s.sessionStore.Close()
	}

	return nil
}

// PublishBinary 向指定topic的所有订阅者发布二进制消息
func (s *Server) PublishBinary(topic string, msg *binary.Message) error {
	// 将二进制消息编码为二进制数据
	data, err := msg.Encode()
	if err != nil {
		return err
	}

	// 使用现有的Publish方法发送消息
	s.Publish(topic, data)
	return nil
}

// SendBinaryToClient 向指定客户端发送二进制消息
func (s *Server) SendBinaryToClient(clientID string, msg *binary.Message) error {
	// 将二进制消息编码为二进制数据
	data, err := msg.Encode()
	if err != nil {
		return err
	}

	// 使用现有的SendToClient方法发送消息
	return s.SendToClient(clientID, data)
}

// SendBinaryToClients 向多个客户端发送相同的二进制消息
func (s *Server) SendBinaryToClients(clientIDs []string, msg *binary.Message) error {
	// 将二进制消息编码为二进制数据
	data, err := msg.Encode()
	if err != nil {
		return err
	}

	// 使用现有的SendToClients方法发送消息
	s.SendToClients(clientIDs, data)
	return nil
}
